package com.jerome.fs.server.aggregation.handler;

import com.jerome.fs.entities.WeatherVo;
import com.jerome.fs.http.HttpMessageParser;
import com.jerome.fs.http.HttpMethodEnum;
import com.jerome.fs.http.Request;
import com.jerome.fs.server.aggregation.ContextUtil;
import com.jerome.fs.sync.CenterServerHandler;
import com.jerome.fs.sync.SyncVo;
import com.jerome.fs.storage.ExpiringMap;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

/**
 * AggregationServer MQ Handler
 *
 * @author Jerome
 * @date 2023/9/5
 */
public class AggregationCenterServerHandler implements CenterServerHandler {
    @Override
    public void handle(Socket socket) throws IOException {
        Request httpRequest = HttpMessageParser.parse2request(socket.getInputStream());
        ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
        if (HttpMethodEnum.SYNC.getCode().equals(httpRequest.getMethod())){
            //Full synchronization
            out.writeObject(new SyncVo(ContextUtil.getExpiringMap(), ContextUtil.getLamportClock().getLatestTime(),httpRequest.getMethod()));
        }else if (HttpMethodEnum.INCREMENT_SYNC.getCode().equalsIgnoreCase(httpRequest.getMethod())) {
            //Incremental synchronization
            List<WeatherVo> weatherVos = new ArrayList<>();
            WeatherVo weatherVo;
            while (true){
                weatherVo = ContextUtil.getRingBuffer().offsetNext();
                if (weatherVo==null){
                    break;
                }
                weatherVos.add(weatherVo);
            }
            out.writeObject(new SyncVo(weatherVos, ContextUtil.getLamportClock().getLatestTime(),httpRequest.getMethod()));
        }else if (HttpMethodEnum.REGISTER.getCode().equalsIgnoreCase(httpRequest.getMethod())) {
            //Registration service
            ExpiringMap<String, String> nodeMap = ContextUtil.getNodeMap();
            InetSocketAddress socketAddress = (InetSocketAddress) socket.getRemoteSocketAddress();
            String address = socketAddress.getAddress().getHostAddress();
            String uri = httpRequest.getUri();
            String[] split = uri.split(":");
            nodeMap.put(address+":"+split[1],"");
            nodeMap.refresh();
            out.writeObject(new SyncVo(nodeMap, ContextUtil.getLamportClock().getLatestTime(),httpRequest.getMethod()));
        }else if (HttpMethodEnum.PINGPONG.getCode().equalsIgnoreCase(httpRequest.getMethod())) {
            //heartbeat check
            ExpiringMap<String, String> nodeMap = ContextUtil.getNodeMap();
            nodeMap.refresh();
            out.writeObject(new SyncVo(nodeMap, ContextUtil.getLamportClock().getLatestTime(),httpRequest.getMethod()));
        }
        out.flush();
    }
}
